<?php

declare(strict_types=1);

namespace App\Exchange\Process;

use App\Exchange\Job\SendContractTickerJob;
use Carbon\Carbon;
use Hyperf\Process\AbstractProcess;
use Hyperf\Process\Annotation\Process;
use Hyperf\Redis\Redis;
use Hyperf\WebSocketClient\Client;
use Hyperf\WebSocketClient\ClientFactory;
use Hyperf\WebSocketClient\Frame;

#[Process(name: 'BinanceContractProcess')]
class BinanceContractProcess extends AbstractProcess
{
    public array $pair = [];
    public Client $client;
    public int $start_at;
    public Redis $redis;

    public function stop(): bool
    {
        return $this->client->close();
    }

    public function isEnable($server): bool
    {
        return false;
        return env('APP_ENV') == 'prod';
    }

    public function handle(): void
    {
        $this->redis = container()->get(Redis::class);
        $this->start();
    }

    public function initClient(): void
    {
        $this->start_at = Carbon::now()->getTimestampMs();
        $this->pair = $this->redis->hGetAll('pair');
        $list = [];
        $interval = ['1m', '5m', '15m', '30m', '1h', '4h', '12h', '1d', '1w', '1M'];
        foreach ($this->pair as $v) {
            foreach (explode(',', strtolower($v)) as $symbol) {
                $list[] = $symbol . '@markPrice@1s';//按Symbol的最优挂单信息
                foreach ($interval as $in) $list[] = $symbol . '@kline_' . $in;
                break;
            }
        }
        $host = 'wss://fstream.binance.com/ws/btcusdt@depth@500ms';
        $this->client = container()->get(ClientFactory::class)->create($host, false);
        $subscribe = [
            'method' => 'SUBSCRIBE',
            'params' => array_splice($list, 0, 1020),
            'id' => time()
        ];
        $this->client->push(json_encode($subscribe));
    }

    public function formatData(): void
    {
        while (true) {
            /** @var $msg Frame */
            $msg = $this->client->recv(1);
            if (!$msg instanceof Frame) continue;
            $res = json_decode($msg->data, true);
            if (isset($res['e'])) {
                if ($res['e'] == 'kline') {
                    $k = 'ckline_' . strtolower($res['s'] . '_' . $res['k']['i']);
                    $this->redis->publish($k, $msg->data);
                    $this->redis->rPush('ex:' . $k, $msg->data);
                    $this->redis->ltrim('ex:' . $k, -300, -1);
                } elseif ($res['e'] == 'markPriceUpdate') {
                    //标记价格, 现货指数价格, 预估结算价, 资金费率,
                    $this->redis->hSet('markPriceUpdate', $res['s'], implode(',', [$res['p'], $res['i'], $res['P'], $res['r']]));
                    fastapp_queue_push(new SendContractTickerJob($msg->data));
                    if ($res['E'] - $this->start_at > 7200000) {
                        $this->stop();
                        break;
                    }
                }
            }
        }
    }

    public function start(): void
    {
        $this->initClient();
        $this->formatData();
        $this->start();
    }
}
